package xyz.riceball.springcloudstreambinderredis.component;

import cn.hutool.core.date.LocalDateTimeUtil;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RDelayedQueue;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.springframework.integration.handler.AbstractMessageHandler;
import org.springframework.messaging.Message;
import xyz.riceball.springcloudstreambinderredis.common.Constants;
import xyz.riceball.springcloudstreambinderredis.common.MessageDTO;

import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.HashMap;
import java.util.concurrent.TimeUnit;

/**
 * <p>
 * RedisMessageHandler
 * </p>
 *
 * @author xiaovcloud
 * @since 2024/6/24 16:49
 */
public class RedisMessageHandler extends AbstractMessageHandler {

	private final String destination;
	private final RedissonClient redissonClient;

	public RedisMessageHandler(String destination, RedissonClient redissonClient) {
		this.destination = destination;
		this.redissonClient = redissonClient;
	}

	@Override
	protected void handleMessageInternal(Message<?> message) {
		MessageDTO<Object> messageDTO = new MessageDTO<>();
		messageDTO.setHeaders(new HashMap<>(message.getHeaders()));
		messageDTO.setPayload(message.getPayload());

		if (messageDTO.getHeaders().containsKey(Constants.DELAY_MESSAGE)) {
			RBlockingQueue<Object> blockingQueue = redissonClient.getBlockingQueue(destination);
			RDelayedQueue<Object> delayedQueue = redissonClient.getDelayedQueue(blockingQueue);
			Object delayTime = messageDTO.getHeaders().get(Constants.DELAY_MESSAGE);
			int delayPeriod = Integer.parseInt((String) delayTime);
			LocalDateTime offset = LocalDateTimeUtil.offset(LocalDateTime.now(), delayPeriod, ChronoUnit.MILLIS);
			messageDTO.getHeaders().put(Constants.DELAY_MESSAGE_TIME, offset);
			delayedQueue.offer(messageDTO, delayPeriod, TimeUnit.MILLISECONDS);
		} else {
			RTopic topic = redissonClient.getTopic(destination);
			topic.publish(messageDTO);
		}
	}
}
